package com.qf.flinketl.sink;

import com.qf.flinketl.mapper.MessageRow;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;

import java.io.Serial;

/**
 * @Description: @Author: KyrieFc @Date: 2023/10/22
 */
public class CustomSink extends RichSinkFunction<MessageRow> {
  @Serial private static final long serialVersionUID = 1L;
  private static final Logger logger = LoggerFactory.getLogger(CustomSink.class);
  private Jedis jedis;

  @Override
  public void open(Configuration parameters) throws Exception {
    super.open(parameters);
    new Jedis("localhost", 6379);
  }

  @Override
  public void invoke(MessageRow value, Context context) throws Exception {
    super.invoke(value, context);
    jedis.lpush("flink", value.getMessage());
  }

  @Override
  public void close() throws Exception {
    super.close();
    if (jedis != null) {
      jedis.close();
    }
  }
}
